package com.enlorenz.core.channel.activemq.common;

import java.util.ArrayList;
import java.util.List;

import com.enlorenz.common.util.TimeUtils;
import com.enlorenz.core.channel.activemq.MQClientChannel;
import com.enlorenz.core.channel.activemq.MQServerChannel;
import com.enlorenz.core.channel.common.message.IDataMessage;
import com.enlorenz.core.channel.common.message.imp.DataMessage;
import com.enlorenz.core.engine.IEngine;
import com.enlorenz.core.engine.manager.EngineManager;
import com.enlorenz.core.queue.transfer.imp.WaitMessageQueue;

/**
 * MQ收发服务
 * @author lcb
 */
public class MQClient2Server {
	 //MQ服务地址
	private String mqUrl;
	//创建通道类型
	private String mqName;
	private WaitMessageQueue sendQueue;
	private MQClientChannel clientChannel;
	private MQServerChannel serverChannel;
	private WaitMessageQueue recvQueue;
	private EngineManager engineManager;
	private Integer limitQueueSize;
	//模式
	private String model;
	
	public static final String TRANS_TYPE_ACTIVEMQ="activemq";
	public static final String TRANS_TYPE_QUEUE="queue";
	
	public void setLimitQueueSize(Integer limitQueueSize) {
		this.limitQueueSize = limitQueueSize;
	}

	public void setRecvQueue(WaitMessageQueue recvQueue) {
		this.recvQueue = recvQueue;
	}
	

	public MQClient2Server() {
		
	}
	
	public void start() throws Exception{
		recvQueue=new WaitMessageQueue();
		recvQueue.setLimitSize(limitQueueSize);
		
		if(model.equals(TRANS_TYPE_QUEUE)){
			return;
		}
		
		List<IEngine> list=new ArrayList<IEngine>();
		serverChannel=new MQServerChannel();
		serverChannel.setName("MQRecv");
		serverChannel.setMqName(mqName);
		serverChannel.setMqUrl(mqUrl);
		
		serverChannel.setRecvQueue(recvQueue);
		list.add(serverChannel);
		
		clientChannel=new MQClientChannel();
		clientChannel.setMqName(mqName);
		clientChannel.setMqUrl(mqUrl);
		clientChannel.setName("MQSend");
		sendQueue=new WaitMessageQueue();
		sendQueue.setLimitSize(limitQueueSize);
		clientChannel.setWaitMessageQueue(sendQueue);
		
		list.add(clientChannel);
		
		
		engineManager=new EngineManager();
		
		
		engineManager.setEngines(list);
		engineManager.setCheckThreadSleepTime(TimeUtils.SECOND_MILLI*2);
		engineManager.startEngines();
	}
	
	/**
	 * 获取传输的类型
	 * @return
	 */
	public String getTransModel(){
		if(model.equals(TRANS_TYPE_QUEUE)){
			return TRANS_TYPE_QUEUE;
		}
		
		if(serverChannel.isAlive()){
			return TRANS_TYPE_ACTIVEMQ;
		}else{
			return TRANS_TYPE_QUEUE;
		}
	}
	
	/**
	 * 发送对象
	 * @param object
	 */
	public void sendObject(Object object){
		IDataMessage om=new DataMessage();
		om.setData(object);
		
		if(model.equals(TRANS_TYPE_QUEUE)){
			recvQueue.addMsg(om);
		}
		
		if(serverChannel.isAlive()){
			sendQueue.addMsg(om);
		}else{
			recvQueue.addMsg(om);
		}
	}
	
	/**
	 * 接收对象
	 * @return
	 */
	public Object recvObject(){
		IDataMessage msg=(IDataMessage)recvQueue.getMsg();
		return msg.getData();
	}
	
	
	public void setMqUrl(String mqUrl) {
		this.mqUrl = mqUrl;
	}

	public void setMqName(String mqName) {
		this.mqName = mqName;
	}
	
	public static void main(String[] args) throws Exception {
		final MQClient2Server ms=new MQClient2Server();
		ms.setMqName("adc01");
		ms.setMqUrl("tcp://localhost:61616");
		ms.setLimitQueueSize(2000);
		ms.start();
		
		new Thread(){
			public void run(){
				while(true){
					Object message=ms.recvObject();
					System.out.println(message);
				}
			}
		}.start();
		
		int index=0;
		while(true){
			try{
				ms.sendObject(index++);
				System.out.println(ms.getTransModel());
				Thread.sleep(200);
			}catch (Exception e) {
				e.printStackTrace();
			}
		}
	}

	public String getModel() {
		return model;
	}

	public void setModel(String model) {
		this.model = model;
	}

}
